package cn.doitedu.udfs;

import cn.doitedu.pojo.*;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

public class MatchFunctionV3 extends KeyedProcessFunction<String, LogBean, MatchResult> {

    private Connection hbaseConn;
    private java.sql.Connection clickHouseConn;
    private RuleConditionV2 ruleCondition;

    @Override
    public void open(Configuration parameters) throws Exception {
        //在TaskManager初始化subtask时，获取全局参数
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        //初始化连接
        //1.创建Hbase的连接
        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", parameterTool.getRequired("hbase.zookeeper.quorum"));
        hbaseConf.setInt("hbase.zookeeper.property.clientPort", parameterTool.getInt("hbase.zookeeper.property.clientPort", 2181));
        hbaseConn = ConnectionFactory.createConnection(hbaseConf);

        clickHouseConn = DriverManager.getConnection(parameterTool.getRequired("clickhouse.url"));

        //创建一个RuleCondition，以后使用广播状态动态广播获取
        ruleCondition = new RuleConditionV2();
        //模拟一个规则，以后是通过广播状态动态接入
        //运营人员，事先定义好的规则
        //触发事件参数
        EventParam eventParam = new EventParam();
        eventParam.setEventId("G");
        HashMap<String, String> props = new HashMap<>();
        props.put("p4", "v1");
        props.put("p9", "v1");
        eventParam.setProps(props);
        ruleCondition.setTriggerEvent(eventParam);
        //用户画像参数
        HashMap<String, String> userProfiles = new HashMap<>();
        userProfiles.put("tag1", "v2");
        userProfiles.put("tag3", "v2");
        ruleCondition.setUserProfiles(userProfiles);
        //行为次数参数
        CombineCondition eventCountCondition = new CombineCondition();

        eventCountCondition.setStartTime(1649433600000L);
        eventCountCondition.setEndTime(1649606399000L);
        eventCountCondition.setCount(2);
        eventCountCondition.setRegex("(1)");
        eventCountCondition.setSql("select eventId from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ?");

        EventParam eventCountParam = new EventParam();
        //事件ID
        eventCountParam.setEventId("G");
        //行为参数
        HashMap<String, String> eventCountProps = new HashMap<>();
        eventCountProps.put("p4", "v1");
        eventCountParam.setProps(eventCountProps);

        eventCountCondition.setEvents(Arrays.asList(eventCountParam));

        //行为序列参数
        CombineCondition eventSeqCondition = new CombineCondition();

        eventSeqCondition.setStartTime(1649433600000L);
        eventSeqCondition.setEndTime(1649606399000L);
        eventSeqCondition.setCount(1);
        eventSeqCondition.setRegex("(.*?1.*?2)");
        eventSeqCondition.setSql("select eventId from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ?");

        //匹配的行为序列A -> H
        EventParam eventParam1 = new EventParam();
        eventParam1.setEventId("A");
        EventParam eventParam2 = new EventParam();
        eventParam2.setEventId("H");

        eventSeqCondition.setEvents(Arrays.asList(eventParam1, eventParam2));

        ruleCondition.setCombineConditions(Arrays.asList(eventCountCondition, eventSeqCondition));
    }

    @Override
    public void processElement(LogBean bean, Context ctx, Collector<MatchResult> out) throws Exception {


        //1.匹配触发事件
        boolean matchEvent = isMatchEvent(bean, ruleCondition.getTriggerEvent()); //行为ID和行为属性
        if (!matchEvent) {
            return;
        }
        //2.匹配用户画像
        boolean matchProfile = isMatchProfile(bean, ruleCondition.getUserProfiles()); //用户画像数据
        if (!matchProfile) {
            return;
        }
        //3.匹配行为次数
        boolean matchEventCount = isMatchEventCount(bean, ruleCondition.getCombineConditions()); //历史行为次数
        if (!matchEventCount) {
            return;
        }

        out.collect(new MatchResult(bean.getDeviceId(), bean.getTimeStamp(), "rule001", System.currentTimeMillis()));

    }

    private boolean isMatchEvent(LogBean bean, EventParam eventParam) {
        //完成匹配触发事件的逻辑
        //eventParam中封装着事先定义好的触发事件ID和属性
        String eventId = eventParam.getEventId();
        Map<String, String> props = eventParam.getProps();
        //用户的实际的事件ID
        //用户实际的行为属性
        Map<String, String> properties = bean.getProperties();
        if (eventId.equals(bean.getEventId())) {
            //匹配行为属性
            for (Map.Entry<String, String> entry : props.entrySet()) {
                //事先定义好的行为属性名称和值
                String targetKey = entry.getKey();
                String targetVal = entry.getValue();
                //根据事先定义好的属性key，取出实际key的value
                String val = properties.get(targetKey);
                if (!targetVal.equals(val)) {
                    return false;
                }
            }
        } else {
            return false;
        }
        return true;
    }

    //是否匹配画像
    private boolean isMatchProfile(LogBean bean, Map<String, String> userProfiles) throws Exception {

        String deviceId = bean.getDeviceId();
        Table table = hbaseConn.getTable(TableName.valueOf("eagle_profile"));
        try {
            Get get = new Get(Bytes.toBytes(deviceId));
            //添加过滤条件
            for (String key : userProfiles.keySet()) {
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(key));
            }
            Result result = table.get(get);
            if (result != null) {
                for (String tag : userProfiles.keySet()) {
                    String val = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes(tag)));
                    //取出事先定义好的该tag的值
                    String targetVal = userProfiles.get(tag);
                    if (!targetVal.equals(val)) {
                        return false;
                    }
                }
            } else {
                return false;
            }
        } finally {
            table.close();
        }
        return true;
    }

    //查询ClickHouse匹配历史行为次数
    //查询clickhouse就是使用基本的select eventId from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ? and properties['p1'] = v1 and and properties['p4'] = v2
    private boolean isMatchEventCount(LogBean bean, List<CombineCondition> combineConditions) throws Exception {

        for (CombineCondition combineCondition : combineConditions) {
            PreparedStatement preparedStatement = null;
            ResultSet resultSet = null;
            try {
                preparedStatement = clickHouseConn.prepareStatement(combineCondition.getSql());
                preparedStatement.setString(1, bean.getDeviceId());
                preparedStatement.setLong(2, combineCondition.getStartTime());
                preparedStatement.setLong(3, combineCondition.getEndTime());
                //返回的resultSet中，装的是多个行为
                resultSet = preparedStatement.executeQuery();
                //用来装实际从clickhouse中查询出的事件（各种各样的事件，可能很多个）
                ArrayList<String> events = new ArrayList<>();
                while (resultSet.next()) {
                    String eventStr = resultSet.getString(1);
                    events.add(eventStr);
                }
                //我实际关系的即要匹配的事件
                List<String> targetEvents = combineCondition.getEvents().stream().map(EventParam::getEventId).collect(Collectors.toList());
                //将实际查出来的事件与要匹配的事件完成映射
                StringBuffer sb = new StringBuffer();
                for (String event : events) {
                    int index = targetEvents.indexOf(event);
                    sb.append(index+1);
                }
                //完成正则匹配的功能
                Pattern pattern = Pattern.compile(combineCondition.getRegex());
                Matcher matcher = pattern.matcher(sb.toString());
                int count = 0;
                while (matcher.find()) {
                    count++;
                }
                //与实际查出来的与条件进行比较，小于就返回false
                if(count < combineCondition.getCount()) {
                    return false;
                }
            } finally {
                if (resultSet != null) {
                    resultSet.close();
                }

                if (preparedStatement != null) {
                    preparedStatement.close();
                }
            }
        }
        return true;
    }

    @Override
    public void close() throws Exception {
        //释放资源
        hbaseConn.close();
        clickHouseConn.close();

    }
}
